package com.mks.demo.mqtt;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mks.demo.entity.WkTest;
import com.mks.demo.service.WkTestService;
import com.mks.demo.utils.IdWorkFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * @Description:
 * @Author: haiLong
 * @Date: 2020/1/19 16:10
 */
@Component
public class MqttDataReceive {

    private static final Logger logger = LoggerFactory.getLogger(MqttDataReceive.class);

    @Autowired
    private WkTestService wkTestService;

    @Value("${history}")
    private Long history;

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                //消息负载数据
                MessageHeaders headers = message.getHeaders();
                String topic = (String) headers.get("mqtt_receivedTopic");
                String[] split = topic.split("/");
                String pKey = split[1];
                String sn = split[2];
                byte[] bytes = (byte[]) message.getPayload();
                String msg = new String(bytes);
                JSONObject jsonObject = JSON.parseObject(msg);
                List<WkTest> list = new ArrayList<>();
                if (jsonObject.containsKey("type")) {
                    if (!jsonObject.containsKey("valid")) {
                        System.out.println("召读历史数据:" + msg);
                    }
                } else {
                    System.out.println("主动上报实时数据:" + msg);
                    //查询最后一条记录
                    WkTest wkTest = wkTestService.getLastestRecord();
                    if (wkTest != null) {
                        //判断当前报文时间和记录时间.
                        if (checkIsHistory(jsonObject.getInteger("ts"), wkTest.getTime())) {
                            System.out.println("已离线");
                            //进行数据召读.

                        } else {
                            System.out.println("未离线");
                        }
                    }
                    JSONArray devs = jsonObject.getJSONArray("devs");
                    for (int i = 0; i < devs.size(); i++) {
                        JSONObject jsonObject0 = JSONObject.parseObject(devs.get(i).toString());
                        JSONArray ds = jsonObject0.getJSONArray("d");
                        for (int j = 0; j < ds.size(); j++) {
                            JSONObject jsonObjectd0 = JSONObject.parseObject(ds.get(j).toString());
                            WkTest wk = new WkTest();
                            wk.setId(IdWorkFactory.getRdWork().nextId());
                            wk.setName(jsonObjectd0.getString("m"));
                            wk.setValue(jsonObjectd0.getString("v"));
                            wk.setTime(new Date(jsonObject.getInteger("ts") * 1000L));
                            list.add(wk);
                        }
                    }
                }
                //数据入库
                wkTestService.insertBatch(list);
            }
        };
    }

    /**
     * 判断是否离线
     *
     * @param ts
     * @param date
     * @return
     */
    private boolean checkIsHistory(Integer ts, Date date) {
        return date.getTime() - (ts * 1000L) > history * 60 * 1000;
    }

}
